package com.orderstats.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.text.SimpleDateFormat;
import java.util.Date;

public class RealTimeOrderProducer {
    
    private static final String[] CATEGORIES = {
        "电子产品", "服装鞋帽", "家居用品", "食品饮料", "图书文具", 
        "运动户外", "美妆护肤", "母婴用品", "汽车用品", "医疗保健"
    };
    
    private static final String[] PRODUCTS = {
        "智能手机", "笔记本电脑", "连衣裙", "运动鞋", "沙发", "咖啡", 
        "小说", "篮球", "口红", "奶粉", "车载充电器", "维生素", 
        "耳机", "T恤", "餐具", "茶叶", "钢笔", "瑜伽垫", "面膜", "玩具"
    };
    
    private static final String[] USERS = {
        "user001", "user002", "user003", "user004", "user005",
        "user006", "user007", "user008", "user009", "user010"
    };
    
    private KafkaProducer<String, String> producer;
    private Random random;
    private SimpleDateFormat dateFormat;
    
    public RealTimeOrderProducer() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        
        this.producer = new KafkaProducer<>(props);
        this.random = new Random();
        this.dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    }
    
    public void startProducing() {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
        
        // 每3秒产生1-3个订单
        scheduler.scheduleAtFixedRate(() -> {
            int orderCount = random.nextInt(3) + 1;
            for (int i = 0; i < orderCount; i++) {
                String orderData = generateOrderData();
                ProducerRecord<String, String> record = new ProducerRecord<>("orders", orderData);
                
                producer.send(record, (metadata, exception) -> {
                    if (exception == null) {
                        System.out.println("订单发送成功: " + orderData);
                    } else {
                        System.err.println("订单发送失败: " + exception.getMessage());
                    }
                });
            }
        }, 0, 3, TimeUnit.SECONDS);
        
        System.out.println("🚀 实时订单数据生产者已启动，每3秒产生1-3个订单");
        System.out.println("📊 数据将发送到Kafka Topic: orders");
    }
    
    private String generateOrderData() {
        String category = CATEGORIES[random.nextInt(CATEGORIES.length)];
        String product = PRODUCTS[random.nextInt(PRODUCTS.length)];
        String user = USERS[random.nextInt(USERS.length)];
        int quantity = random.nextInt(5) + 1;
        String orderTime = dateFormat.format(new Date());
        double rating = 1.0 + random.nextDouble() * 4.0; // 1.0-5.0
        String isValid = random.nextDouble() > 0.1 ? "Y" : "N"; // 90%有效订单
        
        // 格式: category,product,user,quantity,orderTime,rating,isValid
        return String.format("%s,%s,%s,%d,%s,%.1f,%s", 
                category, product, user, quantity, orderTime, rating, isValid);
    }
    
    public void shutdown() {
        if (producer != null) {
            producer.close();
        }
    }
    
    public static void main(String[] args) {
        RealTimeOrderProducer producer = new RealTimeOrderProducer();
        
        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("🛑 正在关闭订单生产者...");
            producer.shutdown();
        }));
        
        producer.startProducing();
        
        // 保持程序运行
        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
} 